package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.behavior.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.time.Duration;

@Slf4j
@Configuration
public class HotArticleStreamHandler {
    //定义bean,注入streamsbuilder
    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //按topic接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);
        //聚合流处理   kafkaTemplate.send(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC, JSON.toJSONString(mess));
        stream.map((key, value) -> {
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    //重置消息的key: 1  value: like:2
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
                })
                //按文章id进行聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //自行完成聚合的计算
                .aggregate(new Initializer<String>() {
                               //初始方法,返回值是消息的value
                               @Override
                               public String apply() {
                                   //aggValue:
                                   return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                               }
                           },
                        //真正的聚合操作
                        new Aggregator<String, String, String>() {
                            @Override
                            public String apply(String key, String value, String aggValue) {
                                if (StringUtils.isBlank(value)) {
                                    return aggValue;
                                }
                                String[] aggAry = aggValue.split(",");
                                int col = 0, com = 0, lik = 0, vie = 0;
                                //聚合结果值:
                                //COLLECTION:2,COMMENT:4,LIKES:1,VIEWS:5
                                for (String agg : aggAry) {
                                    String[] behavior = agg.split(":");
                                    switch (UpdateArticleMess.UpdateArticleType.valueOf(behavior[0])) {
                                        case COLLECTION:
                                            col = Integer.parseInt(behavior[1]);
                                            break;
                                        case COMMENT:
                                            com = Integer.parseInt(behavior[1]);
                                            break;
                                        case LIKES:
                                            lik = Integer.parseInt(behavior[1]);
                                            break;
                                        case VIEWS:
                                            vie = Integer.parseInt(behavior[1]);
                                            break;
                                    }
                                }
                                //累加操作
                                String[] valAry = value.split(":");
                                switch (UpdateArticleMess.UpdateArticleType.valueOf(valAry[0])) {
                                    case COLLECTION:
                                        col += Integer.parseInt(valAry[1]);
                                        break;
                                    case COMMENT:
                                        com += Integer.parseInt(valAry[1]);
                                        break;
                                    case LIKES:
                                        lik += Integer.parseInt(valAry[1]);
                                        break;
                                    case VIEWS:
                                        vie += Integer.parseInt(valAry[1]);
                                        break;
                                }
                                String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                                System.out.println("文章的id" + key);
                                //当前时间窗口类的消息处理结果:COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:1
                                System.out.println("当前时间窗口类的消息处理结果:" + formatStr);
                                return formatStr;
                            }
                        }, Materialized.as("hot_article-stream-count-003")).toStream().map((key, value) -> {
                    return new KeyValue<>(key.key().toString(), formatObject(key.key().toString(), value));
                }).to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);
        return stream;
    }

    private String formatObject(String articleId, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));
        String[] valAry = value.split(",");
        for (String val : valAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.parseInt(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.parseInt(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.parseInt(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.parseInt(split[1]));
                    break;
            }
        }
        //聚合消息处理之后的结果为:{"articleId":1654812672081924098,"collect":0,"comment":0,"like":0,"view":1}
        log.info("聚合消息处理之后的结果为:{}", JSON.toJSONString(mess));
        return JSON.toJSONString(mess);
    }
}

